package com.sxzjrj.utils

import com.sxzjrj.config.MBYLBCConfig
import org.apache.kafka.common.TopicPartition
import org.apache.spark.streaming.kafka010.OffsetRange
import scalikejdbc.{DB, SQL}
import scalikejdbc.config.DBs

/**
  * Created by ljj in 2019/3/25
  *
  */
object OffsetUtils {
  DBs.setup()

  def saveOffset2Mysql(offsetRanges: Array[OffsetRange]) = {

    DB.localTx {
      implicit session =>
        offsetRanges.foreach(offsetRange => {
          SQL(
            """
              |insert into t_offset_mbylbc value(?,?,?,?)
            """.stripMargin)
            .bind(offsetRange.topic,offsetRange.partition,offsetRange.untilOffset,MBYLBCConfig.groupId)
            .update()
            .apply()
        })

    }

  }

  def getCurrentOffset4Mysql() = {
    DB.readOnly {
      implicit session =>

        SQL(
          """
            |select * from t_offset_mbylbc where topics = ? and groupId = ?
          """.stripMargin)
            .bind(MBYLBCConfig.topic,MBYLBCConfig.groupId)
          .map(rs => {
            (new TopicPartition(rs.string("topics"),rs.int("partition")),
              rs.long("untilOffset"))
          }).list().apply()
          .toMap

    }
  }

}
